/*
 * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.processor.impl;

import java.util.Collection;

import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBPersistentStore;
import org.dna.mqtt.moquette.messaging.spi.impl.ValueEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author puran
 */
public class SubscribeProcessor extends AbstractProtocolProcessor {
	private static final Logger LOG = LoggerFactory.getLogger(SubscribeProcessor.class);
	
	public SubscribeProcessor(ServerChannel session, SubscribeMessage message) {
		super(session, message);
	}

	@Override
	public void processEvent(ValueEvent valueEvent) {
		ServerChannel session = getSession(); 
    	SubscribeMessage msg = (SubscribeMessage)getMessage();
    	
		String clientID = (String)session.getAttribute(Constants.ATTR_CLIENTID);
        boolean cleanSession = (Boolean)session.getAttribute(Constants.CLEAN_SESSION);
        
    	LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID());

        for (SubscribeMessage.Couple req : msg.subscriptions()) {
            AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
            Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession);
            
            subscribeSingleTopic(newSubscription, req.getTopic());
        }
        
        //ack the client
        SubAckMessage ackMessage = new SubAckMessage();
        ackMessage.setMessageID(msg.getMessageID());

        //reply with requested qos
        for(SubscribeMessage.Couple req : msg.subscriptions()) {
            AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
            ackMessage.addType(qos);
        }
        
        LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID());
        
        //TODO: why it is writing directly to a session here?
        session.write(ackMessage);
	}
	
	private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
		LOG.info("<{}> subscribed to topic <{}> with QoS {}", newSubscription.getClientId(), topic, AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
        
		MqttProtocolProcessor baseProcessor = getBaseProcessor();
		baseProcessor.addSubscription(newSubscription, newSubscription.getClientId());

        //scans retained messages to be published to the new subscription
        Collection<HawtDBPersistentStore.StoredMessage> messages = baseProcessor.getRetainedMessagesForTopic(topic);
        LOG.debug("Found "+messages.size()+" retained topic message for topic:"+topic);
        
        for (HawtDBPersistentStore.StoredMessage storedMsg : messages) {
            //fire the as retained the message
            LOG.debug("send publish message for topic {}", topic);
            baseProcessor.sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
        }
    }
}
